
#include "EventLoop.h"

#include <iostream>
#include <unistd.h>
#include <sys/eventfd.h>

using std::cerr;
using std::cout;

EventLoop::EventLoop(Acceptor &acceptor)
    : _epollfd(createEpollFd())
    , _evtList()
    , _acceptor(acceptor)
    , _isLooping(false)
    , _wakefd(createEventFd())
{
}

EventLoop::~EventLoop()
{
    ::close(_epollfd);
}
void EventLoop::loop()
{
    _isLooping = true;
    while(_isLooping)
    {
        waitEpollFd();
    }
}

void EventLoop::unloop()
{
    _isLooping = false;
}

int EventLoop::createEpollFd()
{
    int fd = epoll_create1(0);
    if(fd < 0)
    {
        perror("epoll_create1");
    }
    return fd;
}

void EventLoop::addEpollReadFd(int fd)
{
    struct epoll_event evt;
    evt.data.fd = fd;
    evt.events = EPOLLIN;
    if(epoll_ctl(_epollfd, EPOLL_CTL_ADD, fd, &evt) < 0)
    {
        cerr << "addEpollReadFd error\n";
    }
}

void EventLoop::delEpollReadFd(int fd)
{
    struct epoll_event evt;
    evt.data.fd = fd;
    evt.events = EPOLLIN;
    if(epoll_ctl(_epollfd, EPOLL_CTL_DEL, fd, &evt) < 0)
    {
        cerr << "delEpollReadFd error\n";
    }
}

void EventLoop::waitEpollFd()
{
    int nready = 0;
    do {
        nready = epoll_wait(_epollfd, _evtList.data(), _evtList.size(), 2000);
    } while(nready == -1 && errno == EINTR); // 进程收到中断信号 重新调用epoll_wait

    if (nready < 0)
    {
        cerr << "waitEpollFd error\n";
        return;
    }
    else if(nready == 0)
    {
        cout << ">>epoll_wait TIMEOUT.\n";
    }
    else 
    {
        //扩容 ?
        if(nready == static_cast<int>(_evtList.size()))
        {
            _evtList.resize(2 * _evtList.size());
        }

        // 处理就绪事件
        for(size_t i = 0; i < _evtList.size(); ++i)
        {
            int fd = _evtList[i].data.fd;
            // 新连接?
            if(fd == _acceptor.fd())
            {
                if(_evtList[i].events & EPOLLIN)
                {
                    handleNewConnection();
                }
            }
            else if(fd == _wakefd)
            {
                handleRead(); // 重置eventfd : 读出内核计数
                doPengdingFunctors(); // 执行所有任务
            }
            else // old fd
            {
                if (_evtList[i].events & EPOLLIN)
                {
                    handleMessage(fd);
                }
            }
        }
    }
}

void EventLoop::handleNewConnection()
{
    int connfd = _acceptor.accept();
    if(connfd < 0)
    {
        cerr << "handleNewConnection error\n";
        return;
    }

    addEpollReadFd(connfd);
    TcpConnectionPtr pconn = std::make_shared<TcpConnection>(connfd, this); // 创建新连接
    // 注册三个事件
    pconn->setNewConnectionCallback(_onNewConnection);
    pconn->setMessageCallback(_onMessage);
    pconn->setCloseCallback(_onClose);
    _conns[connfd] = pconn;

    // call new connection callback 
    pconn->handleNewConnection();
}
void EventLoop::handleMessage(int fd)
{
    // find conn
    auto it = _conns.find(fd);
    if(it != _conns.end())
    {
        bool isClose = it->second->isClose();
        if(!isClose) // conn open
        {
            //call msg callback
            it->second->handleMessage();
        }
        else // conn close
        {
            // call close callback
            it->second->handleClose();
            // remove fd from epoll and connMap
            delEpollReadFd(fd);
            _conns.erase(it);
        }
    }
    else
    {
        cerr << "TcpConnection not exist\n";
        return;
    }
}

void EventLoop::setNewConnectionCallback(TcpConnectionCallback &&cb)
{
    _onNewConnection = std::move(cb);
}

void EventLoop::setMessageCallback(TcpConnectionCallback &&cb)
{
    _onMessage = std::move(cb);
}

void EventLoop::setCloseCallback(TcpConnectionCallback &&cb)
{
    _onClose = std::move(cb);
}

int EventLoop::createEventFd()
{
    int fd = eventfd(0, 0);
    if(fd < 0)
    {
        cerr << "createEventFd\n";
    }
    return fd;
}

void EventLoop::handleRead()
{
    uint64_t one = 1;
    ssize_t ret = read(_wakefd, &one, sizeof(uint64_t));
    if(ret < 0)
    {
        cerr << "handleRead\n";
    }
}

void EventLoop::doPengdingFunctors()
{
    // 多线程互斥取任务
    vector<Functor> tmp; // 一次取完所有任务
    {
        std::lock_guard(_mutex);
        tmp.swap(_pendings);
    }

    for(auto &func : tmp)
    {
        func(); // 执行任务
    }
}

void EventLoop::wakeup()
{
    uint64_t one = 1;
    ssize_t ret = write(_wakefd, &one, sizeof(uint64_t));
    if (ret != sizeof(uint64_t))
    {
        cerr << "wakeup\n";
    }
}

void EventLoop::runInLoop(Functor &&cb)
{
    // 添加任务
    {
        std::lock_guard(_mutex);
        _pendings.push_back(std::move(cb));
    }
    wakeup(); // 通知执行任务
}